package com.sia.springcloud.protocol.job;

import com.sia.springcloud.model.Device;
import com.sia.springcloud.model.Point;
import com.sia.springcloud.protocol.IProtocol;
import com.sia.springcloud.result.DataCollection;
import lombok.extern.slf4j.Slf4j;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.Date;

/**
 * @Author: Sirius
 * @Date: 2021/12/21 1:42
 * @Description: 数采任务
 **/
@Slf4j
public class CollectJob extends QuartzJobBean {


    @Resource
    private KafkaTemplate kafkaTemplate;


    //数采协议采集类接口，一个数采协议微服务只能有一个实现
    @Resource
    private IProtocol protocol;

    @Override
    protected void executeInternal(JobExecutionContext context) {

        JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
        Device device = (Device) jobDataMap.get("device");
        Point point = (Point) jobDataMap.get("point");
        String topicName = (String) jobDataMap.get("topicName");
        //目前的通道是：Test001

        //设置一个包装类，来存放采集信息
        String value = null;
        DataCollection dataCollection = new DataCollection();
        dataCollection.setDeviceId(device.getId());
        dataCollection.setDeviceName(device.getDeviceName());
        dataCollection.setPointId(point.getId());
        dataCollection.setPointName(point.getPointName());
        dataCollection.setDate(new Date());

        dataCollection.setSuccess(true);


        //获取采集状态和数采值
        try {
            value = this.protocol.read(device, point);
        } catch (Exception e) {
            log.info("数采过程中发生异常：{}", e.getMessage());
            //如果没有发生异常，将会被跳过，也就是不会被设置为false，也即数采表示为true
            dataCollection.setSuccess(false);
            value = "Error";
        } finally {
            dataCollection.setValue(value);
        }

        this.kafkaTemplate.send(topicName, dataCollection).addCallback(
                new ListenableFutureCallback<SendResult<String, Object>>() {
                    @Override
                    public void onFailure(Throwable ex) {
                        log.error("变量名称：" + "xia" + ",发送到kafka失败，原因：" + ex.getMessage());
                    }

                    @Override
                    public void onSuccess(SendResult<String, Object> result) {
                        log.info("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                                + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
                    }
                });

        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }


}
